package com.flinkjar.flinkjardemo.common;


import com.alibaba.fastjson.JSONObject;
import com.flinkjar.flinkjardemo.bean.UserProduct;
import com.flinkjar.flinkjardemo.config.DBConfig;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;


/**
 * 所有表共享一个缓冲区
 */
@Deprecated
public class CustomBatchSinkFunction extends RichSinkFunction<UserProduct> {

    /**
     * 日志
     */
    private static final Logger logger = LoggerFactory.getLogger(CustomBatchSinkFunction.class);

    /**
     * 数据库连接池
     */
    private HikariDataSource dataSource;

    /**
     * 分表数量
     */
    private Integer tableCount = DBConfig.tableCount;

    /**
     * 批处理-缓冲区
     */
    private List<UserProduct> buffer = new ArrayList<>();

    /**
     * 批处理大小
     */
    private int batchSize = 1000;


    /**
     * @param batchSize  批处理大小
     * @param tableCount 分表数量
     */
    public CustomBatchSinkFunction(int batchSize, int tableCount) {
        this.batchSize = batchSize;
        this.buffer = new ArrayList<>();
        this.tableCount = tableCount;
    }


    /**
     * open() 方法会在每个并行的 Sink 算子实例初始化时被调用一次。具体来说，当 Flink 作业启动时，每个 Sink 算子的实例都会被创建并初始化，这时 open() 方法会被调用。
     * 如果 Sink 算子的并行度为 n，则会创建 n 个并行的 Sink 算子实例，
     * 每个实例都会调用一次 open() 方法。
     * <p>
     * open() 方法用于执行一些初始化操作，如建立到外部系统的连接、创建资源等。
     *
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        logger.info("进入open方法.........");

        // 懒加载初始化连接池
        if (dataSource == null) {
            synchronized (this) {
                if (dataSource == null) {
                    // eg: jdbc:mysql://1.92.97.72:3306/flink_test_db?useAffectedRows=true&useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=Asia/Shanghai&allowMultiQueries=true
                    StringBuilder url = new StringBuilder();
                    // rewriteBatchedStatements=true 可以批量插入
                    url.append("jdbc:mysql://").append(DBConfig.hostname_sink).append(":").append(DBConfig.port_sink).append("/").append(DBConfig.database_sink).append("?useAffectedRows=true&useUnicode=true&characterEncoding=utf-8&useSSL=false&connectionTimeZone=Asia/Shanghai&allowMultiQueries=true&rewriteBatchedStatements=true");
                    HikariConfig config = new HikariConfig();
                    // 指定驱动
                    config.setDriverClassName("com.mysql.cj.jdbc.Driver");
                    config.setJdbcUrl(url.toString());
                    config.setUsername(DBConfig.username_sink);
                    config.setPassword(DBConfig.password_sink);
                    // 当池达到这个大小，并且没有空闲连接可用时，对getConnection()的调用将在超时之前阻塞长达connectionTimeout毫秒
                    config.setMaximumPoolSize(5);
                    config.setConnectionTimeout(2000L);
                    // 此属性控制池中连接的最大生存期。当连接达到此超时时，即使最近使用过，它也将从池中退役。正在使用的连接永远不会退役，只有当它空闲时才会被删除。
                    config.setMaxLifetime(1000L);
                    // 该属性控制HikariCP试图在池中维护的空闲连接的最小数量，包括空闲连接和正在使用的连接。如果空闲连接低于此值，HikariCP将尽最大努力快速有效地恢复它们。
                    config.setMinimumIdle(1);
                    dataSource = new HikariDataSource(config);

                    logger.info("数据库连接池创建成功.........");
                }
            }
        }

    }

    /**
     * 写入数据库的逻辑(将数据攒批后输出,可以调高吞吐量)
     * <p>
     * 这种方式仅适用于全量阶段, 增量阶段要求实时性
     *
     * @param value
     * @param context
     */
    @Override
    public void invoke(UserProduct value, Context context) {
        // 将数据加入到缓冲区
        buffer.add(value);

        // 如果缓冲区达到批处理大小，则执行批处理操作
        if (buffer.size() >= batchSize) {
            flushBuffer();
        }

    }

    /**
     * 加入批处理
     *
     * @param value             不可为空
     * @param preparedStatement 不可为空
     * @throws SQLException
     */
    private void doAddBatch(UserProduct value, PreparedStatement preparedStatement) throws SQLException {
        logger.info("加入批处理, id={}", value.getId());
        // insert into user_product_1 (id, user_id, product_id)  values  (?, ?, ?)
        preparedStatement.setLong(1, value.getId());
        preparedStatement.setString(2, value.getUserId());
        preparedStatement.setString(3, value.getProductId());
        // 批处理
        preparedStatement.addBatch();
    }


    /**
     * 任务结束时关闭资源
     *
     * @throws Exception
     */
    @Override
    public void close() throws Exception {
        super.close();

        // 在关闭sink时，确保将剩余的数据进行处理
        if (!buffer.isEmpty()) {
            logger.info("关闭sink时, 将剩余的数据进行处理, size={}", buffer.size());
            flushBuffer();
        }

        // 关闭连接池
        if (dataSource != null) {
            dataSource.close();
        }

        logger.info("资源关闭成功.........");
    }


    /**
     * 刷新缓存
     * 执行批处理操作，将缓冲区中的数据一次性写入目标系统
     * 例如，可以将数据批量写入数据库或者其他外部系统
     * <p>
     * 注意:这里仅演示新增数据
     *
     * @throws Exception
     */
    private void flushBuffer() {
        try {
            logger.info("批处理size={}", buffer.size());
            // key=tableIndex, value=列表
            Map<Integer, List<UserProduct>> bufferMap = buffer.stream().collect(Collectors.groupingBy(i -> i.getTableIndex(tableCount)));


            try (Connection connection = dataSource.getConnection()) {// 获取数据库连接
                // 禁用事务
                connection.setAutoCommit(true);

                /**
                 * 不同的 tableIndex 需要不同的预编译语句
                 */
                bufferMap.forEach((tableIndex, values) -> {
                    logger.info("批处理数据[index={}],[value={}]", tableIndex, JSONObject.toJSONString(tableIndex));
                    // 获取sql模版
                    String sqlTemplate = SqlTemplateService.getInsertSqlTemplate(tableIndex);

                    try (PreparedStatement preparedStatement = connection.prepareStatement(sqlTemplate)) {// 获取预编译语句

                        for (UserProduct value : values) {
                            // 加入批处理
                            doAddBatch(value, preparedStatement);
                        }

                        // 受影响行数
                        int[] affectedRows = preparedStatement.executeBatch();
                        logger.info("批处理, affectedRows={}", affectedRows);
                    } catch (Exception e) {
                        logger.error("数据写入异常, 异常信息:{}", e.getMessage(), e);
                        e.printStackTrace();
                    }
                });

            } catch (Exception e) {
                logger.error("获取数据库连接异常, 异常信息:{}", e.getMessage(), e);
                e.printStackTrace();
            }

            // 清空缓冲区
            buffer.clear();
        } catch (Exception e) {
            logger.error("flushBuffer异常:{}", e.getMessage(), e);
        }
    }

}
